package com.snbc.hbase;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import javax.swing.table.JTableHeader;

/**
 * @author shichenguang
 * @desciption 描述
 * @date 2019-08-29 15:04
 */
public class HbaseTest {
    /**
     * 配置ss
     */
    static Configuration config = null;
    protected static Connection connection = null;
    protected static Admin admin = null;
    protected static Table table = null;

    @Before
    public void init() throws Exception {
        System.getProperties().setProperty("HADOOP_USER_NAME", "hbase");
        config = HBaseConfiguration.create();// 配置
//        config.set("hbase.zookeeper.quorum", "master,work1,work2");// zookeeper地址
//        config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
                config.set("hbase.zookeeper.quorum", "192.168.188.138");// zookeeper地址
        config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
        connection = ConnectionFactory.createConnection(config);
//        table = connection.getTable(TableName.valueOf("shichenguang"));
    }

    /**
     * 创建一个表
     *
     * @throws Exception
     */
    @Test
    public void createTable() throws Exception {
        // 创建表管理类
        HBaseAdmin admin = new HBaseAdmin(connection); // 建立一个操作数据库对象
        TableName tableName = TableName.valueOf("table5");  // 表名称
        HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
        HColumnDescriptor family = new HColumnDescriptor("ss"); // 列族
        tableDescriptor.addFamily(family);
        admin.createTable(tableDescriptor); //创建表
        admin.close();
    }

    @Test
    @SuppressWarnings("deprecation")
    public void deleteTable() throws IOException {
        HBaseAdmin hBaseAdmin = new HBaseAdmin(config);
        hBaseAdmin.disableTable("shichenguang");
        hBaseAdmin.deleteTable("shichenguang");
        hBaseAdmin.close();
    }


    /**
     * 向hbase中增加数据
     *
     * @throws Exception
     */
    @SuppressWarnings({"deprecation", "resource"})
    @Test
    public void insertData() throws Exception {
        table = connection.getTable(TableName.valueOf("shichenguang"));
        //插入数据
        String row = "20190830";
        Put put = new Put(Bytes.toBytes(row));
        put.addColumn(Bytes.toBytes("testInfo"),Bytes.toBytes("data"),Bytes.toBytes(row));
        table.put(put);
    }

    /**
     *
     * 批量插入
     * @throws IOException
     */
    @SuppressWarnings({"deprecation", "resource"})
    @Test
    public void insertDatas() throws IOException {
        table = connection.getTable(TableName.valueOf("shichenguang"));
        ArrayList<Put> puts = new ArrayList<>();
        for (int i = 0; i < 50; i++) {
            Put put = new Put(Bytes.toBytes("127208535" + i));
            put.add(Bytes.toBytes("testInfo"), Bytes.toBytes("name"), Bytes.toBytes("shichenguang" + i));
            put.add(Bytes.toBytes("testInfo"), Bytes.toBytes("password"), Bytes.toBytes("123" + i));
            puts.add(put);
        }
        table.put(puts);

    }

    /**
     * 修改数据
     *
     * @throws Exception
     */
    @Test
    public void uodateData() throws Exception {
        Put put = new Put(Bytes.toBytes("1234"));
        put.add(Bytes.toBytes("info"), Bytes.toBytes("namessss"), Bytes.toBytes("lisi1234"));
        put.add(Bytes.toBytes("info"), Bytes.toBytes("password"), Bytes.toBytes(1234));
        //插入数据
        table.put(put);
        //提交
//        table.flushCommits();
    }

    /**
     * 删除数据
     *
     * @throws Exception
     */
    @Test
    public void deleteDate() throws Exception {
        Delete delete = new Delete(Bytes.toBytes("20190830"));
        table.delete(delete);
//        table.flushCommits();
    }

    /**
     * 单条查询
     *
     * @throws Exception
     */
    @Test
    public void queryData() throws Exception {
        Get get = new Get(Bytes.toBytes("1234"));
        Result result = table.get(get);
        System.out.println(Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("namessss"))));
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("sex"))));
    }




    /**
     * 全表扫描
     * @throws IOException
     */
    @Test
    public void scanData() throws IOException {
        table = connection.getTable(TableName.valueOf("shichenguang"));  //tableName
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        Map itemMap = new LinkedHashMap();
        scanner.forEach(e->{
            e.list().forEach(item ->{
                printDataUtil(itemMap, item);
            });
        });
    }



    /**
     * 查看指定的列
     * @throws IOException
     */
   @Test
    public void getResult() throws IOException {
        String rowKey = "20190830";
        Get get=new Get(Bytes.toBytes(rowKey));
        //tableName
        table = connection.getTable(TableName.valueOf("shichenguang"));
        get.addColumn(Bytes.toBytes("testInfo"),Bytes.toBytes("name"));
        Result result = table.get(get);
        Map itemMap = new LinkedHashMap();
        result.list().forEach(kv ->{
            printDataUtil(itemMap, kv);
        });
    }
    /**
     * 全表扫描的过滤器
     * 列值过滤器
     *
     * @throws Exception
     */
    @Test
    public void scanDataBySingleColumnValueFilter1() throws Exception {

        // 创建全表扫描的scan
        Scan scan = new Scan();
        //过滤器：列值过滤器
        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("info"),
                Bytes.toBytes("name"), CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes("zhangsan2"));
        // 设置过滤器
        scan.setFilter(filter);

        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            System.out.println(Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))));
            //System.out.println(Bytes.toInt(result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("password"))));
            //System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("name"))));
        }

    }

    /**
     * 过滤器
     * 列值过滤器
     * 注意：如果过滤器过滤的列在数据表中有的行中不存在，那么这个过滤器对此行无法过滤。
     * @throws Exception
     */
    @Test
    public void scanBySingleColumnValueFilter() throws Exception{
        //全表扫描
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("testInfo"),
                Bytes.toBytes("password"), CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes("1239"));
        scan.setFilter(singleColumnValueFilter);
        ResultScanner scanner = table.getScanner(scan);
        LinkedHashMap<Object, Object> map = new LinkedHashMap<>();
        scanner.forEach(e-> e.list().forEach(kv -> printDataUtil(map, kv)));
    }


    /**
     * row 扫描
     * @throws Exception
     */
    @Test
    public void scanByRowFilter() throws Exception{
        Scan scan = new Scan();
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryComparator(Bytes.toBytes("12720853543"))  // 匹配完整字节数组
//                  new BinaryPrefixComparator(Bytes.toBytes("127"))  //匹配字节数组前缀
//                  new SubstringComparator("085")    //子串匹配，相当于 contains()
//                 new RegexStringComparator("^12341") //正则匹配
        );
        scan.setFilter(rowFilter);
        ResultScanner scanner = table.getScanner(scan);
        LinkedHashMap<Object, Object> map = new LinkedHashMap<>();
        scanner.forEach(e->e.list().forEach(keyValue -> printDataUtil(map,keyValue)));
    }

    /**
     * 匹配列名前缀
     *
     * @throws Exception
     */
    @Test
    public void scanDataByFilter3() throws Exception {

        // 创建全表扫描的scan
        Scan scan = new Scan();
        //匹配rowkey以wangsenfeng开头的
        ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("na"));
        // 设置过滤器
        scan.setFilter(filter);
        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            System.out.println("rowkey：" + Bytes.toString(result.getRow()));
            System.out.println("info:name："
                    + Bytes.toString(result.getValue(Bytes.toBytes("info"),
                    Bytes.toBytes("name"))));
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")) != null) {
                System.out.println("info:age："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info"),
                        Bytes.toBytes("age"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info"), Bytes.toBytes("sex")) != null) {
                System.out.println("infi:sex："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info"),
                        Bytes.toBytes("sex"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("name")) != null) {
                System.out
                        .println("info2:name："
                                + Bytes.toString(result.getValue(
                                Bytes.toBytes("info2"),
                                Bytes.toBytes("name"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("age")) != null) {
                System.out.println("info2:age："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info2"),
                        Bytes.toBytes("age"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("sex")) != null) {
                System.out.println("info2:sex："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info2"),
                        Bytes.toBytes("sex"))));
            }
        }

    }

    /**
     *
     * 列名前缀匹配rowkey,查出对应列的值
     * @throws Exception
     */
    @Test
    public void scanColumnPrefixFilter() throws Exception{

        // 创建全表扫描的scan
        Scan scan = new Scan();
        ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("pa"));
        // 设置过滤器
        scan.setFilter(filter);
        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        HashMap<Object, Object> map = new HashMap<>();
        scanner.forEach(e->e.list().forEach(keyValue -> printDataUtil(map,keyValue)));
    }

    /**
     * 过滤器集合
     *
     * @throws Exception
     */
    @Test
    public void scanDataByFilter4() throws Exception {

        // 创建全表扫描的scan
        Scan scan = new Scan();
        //过滤器集合：MUST_PASS_ALL（and）,MUST_PASS_ONE(or)
        FilterList filterList = new FilterList(Operator.MUST_PASS_ONE);
        //匹配rowkey以wangsenfeng开头的
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("^wangsenfeng"));
        //匹配name的值等于wangsenfeng
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                Bytes.toBytes("name"), CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes("zhangsan"));
        filterList.addFilter(filter);
        filterList.addFilter(filter2);
        // 设置过滤器
        scan.setFilter(filterList);
        // 打印结果集
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            System.out.println("rowkey：" + Bytes.toString(result.getRow()));
            System.out.println("info:name："
                    + Bytes.toString(result.getValue(Bytes.toBytes("info"),
                    Bytes.toBytes("name"))));
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")) != null) {
                System.out.println("info:age："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info"),
                        Bytes.toBytes("age"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info"), Bytes.toBytes("sex")) != null) {
                System.out.println("infi:sex："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info"),
                        Bytes.toBytes("sex"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("name")) != null) {
                System.out
                        .println("info2:name："
                                + Bytes.toString(result.getValue(
                                Bytes.toBytes("info2"),
                                Bytes.toBytes("name"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("age")) != null) {
                System.out.println("info2:age："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info2"),
                        Bytes.toBytes("age"))));
            }
            // 判断取出来的值是否为空
            if (result.getValue(Bytes.toBytes("info2"), Bytes.toBytes("sex")) != null) {
                System.out.println("info2:sex："
                        + Bytes.toInt(result.getValue(Bytes.toBytes("info2"),
                        Bytes.toBytes("sex"))));
            }
        }

    }

    //    @After
//    public void close() throws Exception {
//        table.close();
//        connection.close();
//    }
    @After
    public void close() {
        try {
            if (connection != null) {
                connection.close();
            }
            if (admin != null) {
                admin.close();
            }
            if (table != null) {
                table.close();
            }
        } catch (Exception e) {

        }

    }

    /**
     *控制台打印查询出来的数据
     * @param itemMap
     * @param item
     */
    private void printDataUtil(Map itemMap, KeyValue item) {
        itemMap.put("Row", Bytes.toString(item.getRow()));
        itemMap.put("Family",Bytes.toString(item.getFamily()));
        itemMap.put("Qualifier",Bytes.toString(item.getQualifier()));
        itemMap.put("Value",Bytes.toString(item.getValue()));
        System.out.println(itemMap);
    }

}
